package com.aotain.project.apollo.bolt;

import java.text.SimpleDateFormat;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.commons.codec.binary.Base64;
import org.apache.log4j.Logger;
import org.apache.storm.Config;
import org.apache.storm.Constants;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.FailedException;
import org.apache.storm.topology.IRichBolt;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;

import com.aotain.hbase.HBaseRecordAdd;
import com.aotain.apollo.DosSliding;
import com.aotain.apollo.utils.ApolloProperties;
import com.aotain.storm.AbstractBolt;

/**
 * @ClassName: CCDetectBolt 
 * @Description:  CC攻击检测方法，HBASE入库
 * @author Turk
 * @date 2015年  
 * 
 * 2016/11/02 取消连续3分钟判断条件
 */
public class CCDetectBolt extends AbstractBolt{

	/**
	 * 
	 */
	private static final long serialVersionUID = 1L;

	/**
	 * 每分钟累积MAP  目标|源|URL
	 */
	private HashMap<String,Integer> _mapUrl = new HashMap<String,Integer>();
	
	/**
	 * 上行包字节数
	 */
	private HashMap<String,Integer> _mapStreamoctets = new HashMap<String,Integer>();
	
	/**
	 * 上行包数
	 */
	private HashMap<String,Integer> _mapStreampacket = new HashMap<String,Integer>();

	//private HashMap<String,Integer> _mapWUrl = new HashMap<String,Integer>();

	
	/**
	 * 滑动窗口对象
	 */
	private DosSliding cache;
	
	public CCDetectBolt()
	{
		cache = new DosSliding(5);//
	}


	@Override
	public void execute(Tuple tuple) {
		// TODO Auto-generated method stub
		try {
			if(isTickTuple(tuple)) {
				//countInLocal(tuple);
				HashMap<String,Integer> mapUrl = new HashMap<String,Integer>();
				mapUrl.putAll(_mapUrl);
				Logger.getLogger(CCDetectBolt.class).info(String.format("###DEBUG:CC START EMIT [%s]",tuple.getSourceStreamId()));
				emitCountingData(collector, mapUrl);
				Logger.getLogger(CCDetectBolt.class).info(String.format("###DEBUG:CC END EMIT [%s]",tuple.getSourceStreamId()));
				_mapUrl.clear();
				collector.ack(tuple);
			} else {
				countInLocal(tuple); 
				collector.ack(tuple);
			}
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			collector.fail(tuple);
			throw new FailedException("CCDetectBolt throws an exception!",e);
		}
	}
	
	
	//计算需要的值
	private void countInLocal(Tuple tuple) {
		try
		{
			/*//"dip","dport","sip","sport","gisstr","domainname","url"*/
			String url = tuple.getStringByField("url");	
			if(url==null || url.trim().isEmpty())
				return;
			
			String dip = tuple.getStringByField("dip");
			String sip = tuple.getStringByField("sip");
			String idc = tuple.getStringByField("idcno");
					
			String gis = tuple.getStringByField("gisstr");
			
			String useragent = "useragent";
			String cookie = "cookie";
			if(tuple.contains("useragent") && tuple.contains("cookie"))
			{
				useragent = tuple.getStringByField("useragent");
				cookie = tuple.getStringByField("cookie");
			}
			
			//String gis = new String(Base64.decodeBase64(tuple.getStringByField("gisstr")),"UTF8");			
			String tripleKey = dip + "|" + sip +"|" + url + "|" 
					+ gis + "|" + idc + "|" + useragent + "|" + cookie;
			
			
			
			//发现相同URL，相同源IP，+1
			int count = 1;
			if(_mapUrl.containsKey(tripleKey))
			{
				count = _mapUrl.get(tripleKey) + 1;
			}
			_mapUrl.put(tripleKey, count);
			
			
			//UDII数据特有
			if(tuple.contains("upstreamoctets") && tuple.contains("upstreampacket"))
			{
				int nUpStreamOctets = tuple.getIntegerByField("upstreamoctets");
				int nUpstreampacket = tuple.getIntegerByField("upstreampacket");
				if(_mapStreamoctets.containsKey(tripleKey))
				{
					nUpStreamOctets = _mapStreamoctets.get(tripleKey) + nUpStreamOctets;
				}
				_mapStreamoctets.put(tripleKey, nUpStreamOctets);
				
				if(_mapStreampacket.containsKey(tripleKey))
				{
					nUpstreampacket = _mapStreampacket.get(tripleKey) + nUpstreampacket;
				}
				_mapStreampacket.put(tripleKey, nUpstreampacket);
			
			}
		}
		catch(Exception ex)
		{
			Logger.getRootLogger().error("=====countInLocal=========ERROR=================",ex);
		}
	}
	
	private void emitCountingData(OutputCollector collector,HashMap<String,Integer> mapUrl) {
		
		try
		{
		HBaseRecordAdd hbaseInstance = HBaseRecordAdd.getInstance(zooserver);
		
		SimpleDateFormat sdf  = new SimpleDateFormat("yyyyMMddHHmm");		
		Calendar c = Calendar.getInstance();
		String dataMin = sdf.format(c.getTime());
		c.add(Calendar.MINUTE, -1);// 1 minute ago
		String dateMin1 = sdf.format(c.getTime());
		c.add(Calendar.MINUTE, -1);// 2 minutes ago
		String dateMin2 = sdf.format(c.getTime());
		
//		String info = String.format("######DEBUG:CC MAP[%d] T[%s] T1[%s] T2[%s]", 
//			 mapUrl.size(),	dataMin , dateMin1 , dateMin2);
//		Logger.getRootLogger().info(info);
		
		HashMap<String,Integer> mapWUrl = new HashMap<String,Integer>(); 
		for(Entry<String, Integer> entry : mapUrl.entrySet())
		{
			//当前分钟
			String minKey = entry.getKey() + "|" + dataMin;
			mapWUrl.put(minKey, entry.getValue());
			
//			info = String.format("######DEBUG:CC T[%s] NUM[%d] VALUE[%s]", 
//					dataMin , entry.getValue(), minKey );
//			Logger.getRootLogger().info(info);
			
			if(entry.getValue()>= 1800) //每秒300
			{//同一个源IP 到目的IP 相同的URL 在一分钟内访问次数超过600次 判断是否连续
				//Map<Integer,HashMap<String,Integer>> infoMap = cache.getTuplesWindow();	
				
				
				/* 2016/11/02 取消连续3分钟判断
				if(infoMap != null ) {
					int num = 0;
					for(HashMap<String,Integer> singleMap: infoMap.values()) {
						
						if(singleMap != null) {
							if(singleMap.containsKey(entry.getKey() + "|" + dateMin1) 
									|| singleMap.containsKey(entry.getKey() + "|" + dateMin2)){								
								num++;								
							}
						}
					}
					*/
					
					//if(num >= 2)
					{//满足连续条件
						String[] tuplesp = entry.getKey().split("\\|",-1);
						//System.out.println("target >>>>>>>>> "+tupleMap.get(dkey) + " >>>>>>>>> frequency:" +  tripleMap.get(dkey));
						//dip + "|" + sip +"|" + url + "|" + gis
//						Logger.getRootLogger().info("####DEBUG:CC --OK!!!!####" + entry.getKey());
						String dip = tuplesp[0];
						String sip = tuplesp[1];
						String url = tuplesp[2];
						url = new String(Base64.decodeBase64(url));	
						//url = url.replace(url.substring(0,10), "*");
						String gis = new String(Base64.decodeBase64(tuplesp[3]));
//						Logger.getRootLogger().info("####CC-gis####" + gis);
						//String gis = tuplesp[3];
						String[] gisArray = gis.split("#",-1);
						String idc = tuplesp[4];

						String destAreaName = gisArray[0];
						String destGis = gisArray[1];
						String sourceAreaName = gisArray[2];
						String sourceGis = gisArray[3];
						String sourceAreaCountry = gisArray[4];
						String sourceAreaId = gisArray[5];							
						String sourceProvinceName =  gisArray[6].isEmpty()?sourceAreaCountry:gisArray[6]; //如果省为空，精确到国家
						String sourceAreaCityId = gisArray[7];
						String sourceAreaProvinceId = gisArray[8];
						
						String accesstime = new SimpleDateFormat("yyyyMMddHHmmss").format(new Date(System.currentTimeMillis()+15000));
						String rowKeyAbnormal = dip + "_" + accesstime + "_" + sip + "_" + "_3";	

						String desc = String.format("WEB攻击 ,攻击次数[%d]", entry.getValue());
						if(_mapStreamoctets.size() > 0 && _mapStreampacket.size() > 0)
						{
							int nUpStreamOctets = _mapStreamoctets.get(entry.getKey());
							//int nUpstreampacket = _mapStreampacket.get(entry.getKey());
							desc = String.format("%s,攻击流量[%d KB]", desc, nUpStreamOctets/1024);
						}
						
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEIP", sip);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTPORT", "");							
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ACCESSTIME", accesstime);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ABRNORMAL", "3");
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTIP", dip);							
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEAREA", sourceAreaName);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCEGEO", sourceGis);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "SOURCECOUNTRY", sourceAreaCountry);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTAREA", destAreaName);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESTGEO", destGis);							
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "ATTNUM", entry.getValue().toString());
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "DESC", desc);
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "EVALUATE", "40");
						hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "PROVINCE", sourceProvinceName);
						
						
						String tbName3 = ApolloProperties.SDS_ABNORMAL_LOG3;
				    	String rowkey3 = String.format("%s_%s", accesstime, sip);
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "ACCESSTIME", accesstime);
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEIP", sip);
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCEGEO", String.format("%s",sourceGis));
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "ATTNUM", entry.getValue().toString());
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "AREAID", sourceAreaProvinceId);
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "AREANAME", sourceProvinceName);
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYID", sourceAreaCityId);
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "CITYNAME", sourceAreaName);
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTIP", dip);
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTAREA", destAreaName);
//				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTPORT", "");
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "SOURCECOUNTRY", sourceAreaCountry);
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "TYPE", "2");
				    	hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);
				    	
						Date current = new Date();
						SimpleDateFormat sdf_h  = new SimpleDateFormat("yyyyMMddHH");
						SimpleDateFormat sdf_d  = new SimpleDateFormat("yyyyMMdd");
						SimpleDateFormat sdf_hour  = new SimpleDateFormat("HH:00");
				    	String dateStr_h = sdf_h.format(current);
						String dateStr_d = sdf_d.format(current);
						String dateStr_hour = sdf_hour.format(current);
						
						int countValue = entry.getValue();
						
						//UDII数据特有
						int nUpStreamOctets = 0;
						int nUpstreampacket = 0;
						if(_mapStreamoctets.size() > 0 && _mapStreampacket.size() > 0)
						{
							nUpStreamOctets = _mapStreamoctets.get(entry.getKey());
							nUpstreampacket = _mapStreampacket.get(entry.getKey());
							hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMOCTETS", String.valueOf(nUpStreamOctets));
							hbaseInstance.Add("SDS_ABNORMAL_LOG2", rowKeyAbnormal, "cf", "UPSTREAMPACKET", String.valueOf(nUpstreampacket));
							countValue = nUpStreamOctets;
//							hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);
//							hbaseInstance.Add(tbName3, rowkey3, "cf", "DESTGEO", destGis);
						}
						
						collector.emit(new Values(dip,sip,"","",dateStr_d,dateStr_h,sourceAreaId,sourceAreaName,
				    			sourceProvinceName, countValue, dateStr_hour,"CC", idc, sourceAreaCountry,(long)nUpStreamOctets,(long)nUpstreampacket));
				    	
					}
				}
			}
			
			
		//}
		
//		if(mapWUrl.size() > 0){
//			cache.addPInfo(mapWUrl);
//		}
		
		//_mapUrl.clear();
		_mapStreamoctets.clear();
		_mapStreampacket.clear();
		//cache.advanceHead();
		
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			throw new FailedException("CCDetectBolt emitCountingData throws an exception!",e);
		}
	}
		

	@Override
	public void cleanup() {
		// TODO Auto-generated method stub
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("dip","sip","dport","sport","dateStr_d",
				"dateStr_h","sourceAreaId","sourceAreaName","areaName","pv","dateStr_hour",
				"abntype","idcno","sourcecountry","streamoctets","streampacket"));
		declarer.declareStream("ABNORMAL_LOG", new Fields("rowkey","dip","sip","dport",
				"accesstime","sourcearea","sourcegeo","sourcecountry","destarea","destgeo",
				"evaluate","abnormal","desc","attnum","delaytime","flag"));
		
	}

	@Override
	public Map<String, Object> getComponentConfiguration() {
		// TODO Auto-generated method stub
		Map<String,Object> conf = new HashMap<String,Object>();
		conf.put(Config.TOPOLOGY_TICK_TUPLE_FREQ_SECS,60);
		return conf;
	}


	@Override
	public void Init(Map stormConf, TopologyContext context,
			OutputCollector collector) {
		// TODO Auto-generated method stub
		
	}

}
